package com.gzyj.flink.gps.map;

import com.gzyj.flink.dto.alarm.Kafka_VehicleNormalAlarm;
import com.gzyj.flink.gps.GPSPosition;
import com.gzyj.flink.gps.map.fatiguedriving.FatigueDrivingResult;
import com.gzyj.flink.gps.map.fatiguedriving.TotalFatigueDrivingDataProcess;
import com.gzyj.flink.pojo.gps.BaobiaoAlarmsummary;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import java.util.Objects;


public class TotalFatigueDrivingFlow {
    private SingleOutputStreamOperator<GPSPosition> gpsSingleStream;

    public TotalFatigueDrivingFlow(SingleOutputStreamOperator<GPSPosition> gpsSingleStream) {
        this.gpsSingleStream = gpsSingleStream;
    }

    public DataStream<BaobiaoAlarmsummary> process(){
        SingleOutputStreamOperator<FatigueDrivingResult> fatigueDrivingResultSingleOutputStream= this.gpsSingleStream.flatMap(new TotalFatigueDrivingDataProcess()).filter(Objects::nonNull);

        SingleOutputStreamOperator<Kafka_VehicleNormalAlarm> kafkaVehicleNormalAlarmSingleOutputStream = fatigueDrivingResultSingleOutputStream
                .filter(e -> e.getKafkaVehicleNormalAlarm() != null)
                .map(e -> e.getKafkaVehicleNormalAlarm());

        kafkaVehicleNormalAlarmSingleOutputStream.print();

        DataStream<BaobiaoAlarmsummary> baobiaoAlarmsummaryDataStream= new BeidouAlarmProcess(kafkaVehicleNormalAlarmSingleOutputStream).process();

        fatigueDrivingResultSingleOutputStream
                .filter(e -> e.getKafkaVehicleRemindContent() != null)
                .map(e -> e.getKafkaVehicleRemindContent())
                .print();

        return baobiaoAlarmsummaryDataStream;
    }

}
